Flaskの拡張機能を使ったRESTful APIの実装

機能拡張を使ったRESTful API

FlaskによるWebサービスの実装 Part1 、Part2 では Flask だけを使ってRESTful Webサービスを実装しました。

Flask で RESTful API を実装するための機能拡張には次のようなものがあります。

Flask-RESTful：RESTful/APIを簡単に実装するときに便利

Fask-restx：RESTful/APIやAPIドキュメントの自動化を提供

HTTPメソッドのマッピング

WebサービスTODOの仕様を少し変更しています。

table: APIとHTTPメソッド

HTTPメソッド URI アクション

GET http://hostname/todo/api/v3.0/tasks タスクリストの取得

GET http://hostname/todo/api/v3.0/tasks/id タスクの取得

POST http://hostname/todo/api/v3.0/tasks 新規タスクの作成

PUT http://hostname/todo/api/v3.0/tasks/id 既存タスクを更新

DELETE http://hostname/todo/api/v3.0/tasks/id 指定したタスクを削除

タスクは次の情報を持つものとします。

uri：タスクを示す一意のURI。String型。

title：タスクのタイトル。タスクについての短い説明。 String型。

description：タスクの詳細。タスクについての詳細な説明。 Text型。

done：タスクの完了状態。 Boolean型。

Flask-RETfulでの実装

Flask-RESTful はREST APIの構築を容易するためのFlaskの拡張機能です。

Flask-RESTful の主な機能には次のものがあげられます。

リクエストデータの解析

リクエストデータの解析reqparse.RequestParser()は、Pythonのコマンドラインオプション解析パーサargparserのインターフェイスに準拠しています。RESTfulリクエストパーサーはリクエストオブジェクト(flask.request）にあるすべての変数に同じ方法で直接にアクセスできるようになります。

出力フィールド

ほとんどのアプリケーション開発者は応答データのレンダリングを制御することを好みます。Flask-RESTfulは、ORMモデルやカスタムクラスについてもレンダリングするオブジェクトとして使用できるメカニズムを提供しています。

Flask-RESTful では応答オブジェクトはJSONなど指定した形式でフィルタリングされます。

アプリケーション開発者は内部データ構造が公開される心配をしなくてもすみます。

ここで、Flask-RETful を使用してWebサービスを実装しみましょう。

この資料は、Designing a RESTful API with Python and Flask をもとに作成されています。

インストール

Flask-RESTfulをインストールします。

code: bash

$ pip install Flask-RESTful

準備

アプリケーションのディレクトリを作成します。

code: bash

$ mkdir -p $HOME/flask/todo_apiv3

$ cd $HOME/flask/todo_apiv3

Flask-RESTful のResourceクラス

Flask-RESTfulは、指定されたURIに1つ以上のHTTPメソッドをルーティング設定できるResourceクラスを提供します。たとえば、TaskAPIリソースでHTTPプロトコルのGET、PUT、DELETEメソッドを定義するには、次のように記述します。

code: python

class TaskAPI(Resource):

def get(self, id):

pass

def put(self, id):

pass

def delete(self, id):

pass

api.add_resource(TaskAPI,

'/todo/api/v3.0/tasks/<int:id>', endpoint = 'task')

関数add_resource()は、指定されたエンドポイントを使用して、ルーティングをフレームワークに登録します。エンドポイントを省略すると Flask-RESTful はクラス名からエンドポイントを生成しますが、url_for()などでエンドポイントが必要になる場合があるため、明示的に指定するようにします。

Webサービス TODOのAPIではGETメソッドは2つのURIを使用します。

タスク一覧を取得するための /todo/api/v3.0/tasks

特定のタスクを取得する/todo/api/v3.0/tasks/<int:id>

Flask-RESTfulのResourceクラスはひとつのURIしかルーティング設定できないので、このWebサービスには2つのリソースクラスが必要になります。

タスクを登録するときのPOST、タスクを更新するPUTではリクエストデータのフィールドの省略ができるかどうかが異なるので、この２つのリソースクラスに分けて設定します。

code: python

class TaskListAPI(Resource):

def get(self):

pass

def post(self):

pass

class TaskAPI(Resource):

def get(self, id):

pass

def put(self, id):

pass

def delete(self, id):

pass

api.add_resource(TaskListAPI,

'/todo/api/v3.0/tasks', endpoint = 'tasks')

api.add_resource(TaskAPI,

'/todo/api/v3.0/tasks/<int:id>', endpoint = 'task')

TaskListAPIクラスのメソッドは引数を受け取りませんが、TaskAPIクラスのすべてのメソッドはURIで指定されているIDを引数として受け取るようにしておきます。

リクエストの解析と検証

FlaskによるWebサービスの実装 Part1では、リクエストされたデータの解析と検証を独自に実装していました。

code: python

@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods='PUT')

def update_task(task_id):

task = [task for task in tasks if task'id' == task_id]

if len(task) == 0:

abort(404)

if not request.json:

abort(400)

if ('title' in request.json and

isinstance(type(request.json'title'), unicode)):

abort(400)

if ('description' in request.json and

isinstance(type(request.json'description'), unicode)):

abort(400)

if ('done' in request.json and

type(request.json'done') is not bool):

abort(400)

task0'title' = request.json.get('title', task0'title')

task0'description' = \

request.json.get('description', task0'description')

task0'done' = request.json.get('done', task0'done')

return jsonify({'task': make_public_task(task0)})

リクエストで指定されたデータが有効であることを確認するために、リソースのフィールドが多ければ多いほど関数が長くなってしまいます。

Flask-RESTfulの、RequestParserクラスを使うと、もっと簡単に記述することができるようになります。このクラスは、コマンドライン引数の解析モジュール argparse と同じような使い方で設定します。

code: python

from flask_restful import reqparse

class TaskListAPI(Resource):

def __init__(self):

self.reqparse = reqparse.RequestParser()

self.reqparse.add_argument('title',

type = str, required = True,

help = 'No task title provided',

location = 'json')

self.reqparse.add_argument('description',

type = str, default = "",

location = 'json')

super(TaskListAPI, self).__init__()

def get(self):

pass

def post(self):

pass

class TaskAPI(Resource):

def __init__(self):

self.reqparse = reqparse.RequestParser()

self.reqparse.add_argument('title',

type = str, location = 'json')

self.reqparse.add_argument('description',

type = str, location = 'json')

self.reqparse.add_argument('done',

type = bool, location = 'json')

super(TaskAPI, self).__init__()

def get(self, id):

pass

def put(self, id):

pass

def delete(self, id):

pass

api.add_resource(TaskListAPI,

'/todo/api/v3.0/tasks', endpoint = 'tasks')

api.add_resource(TaskAPI,

'/todo/api/v3.0/tasks/<int:id>', endpoint = 'task')

TaskListAPIリソースクラスでは、リクエストデータを解析する必要があるのはpost()メソッドのときだけです。RequestParserクラスが読み出すリクエストデータの場所をlocation='json'で指定します。これにより、request.jsonを解析するようになります。

ここではtitleフィールドは必須(required = True)で、指定されていない場合のクライアントへの応答として送信するエラーメッセージをhelp="..."で指定しています。descriptionフィールドは省可能で、デフォルト値（default = "")を指定しています。

TaskAPIリソースクラスは同様の方法で作成されていますが、引数を解析する必要があるのはput()メソッドになります。

ここではすべてのフィールドは省略可能としています。(requiredのデフォルトは False）

リクエストパーサーが初期化されると、リクエストの解析と検証はとても簡単になります。

TaskAPIリソースクラスのput()メソッドは次のように self.reqparse.parse_args()を呼び出すだけですみます。

code: python

class TaskAPI(Resource):

# ...

def put(self, id):

task = filter(lambda t: t'id' == id, tasks)

if len(task) == 0:

abort(404)

task = task0

args = self.reqparse.parse_args()

for k, v in args.iteritems():

if v != None:

taskk = v

return jsonify( { 'task': make_public_task(task) } )

Flask-RESTful を使う利点は、リクエストを解析/検証したときの、不正な要求コード400や404などに対してエラーハンドラーの定義が不要になることです

応答の生成

"FlaskによるWebサービスの実装 Part1"では、Flaskのjsonify()関数を使用して応答を生成しました。

code: python

return jsonify( { 'task': make_public_task(task) } )

Flask-RESTful ではJSONへの変換を自動的に処理してくれます。

code: python

return { 'task': make_public_task(task) }

Flask-RESTful はステータスの指定もサポートしています。

code: python

return { 'task': make_public_task(task) }, 201

"FlaskによるWebサービスの実装 Part1"では make_public_task()を、

タスクを内部表現からクライアントが期待する外部表現に変換していました。

(id フィールドに変えてuriフィールドを追加することなど)

Flask-RESTful は、URIを生成するだけでなく、残りのフィールドで型変換も行ってくれる関数marshal()が提供されています。

code: python

from flask_restful import fields, marshal

task_fields = {

'title': fields.String,

'description': fields.String,

'done': fields.Boolean,

'uri': fields.Url('task')

}

class TaskAPI(Resource):

# ...

def put(self, id):

# ...

return { 'task': marshal(task, task_fields) }

task_fieldsは、整列化関数marshal()のテンプレートとして機能します。 fields.Url()は、URLを生成する特別なタイプで、エンドポイントを引数に与えます。

ここまでのまとめ

コードを整理してみましょう。

code: conf.py

allow_users = {

'freddie': {

'username': 'freddie',

'password': 'queen'

}

code: tasks.py

from flask_restful import fields

task_fields = {

'title': fields.String,

'description': fields.String,

'done': fields.Boolean,

'uri': fields.Url('task')

}

tasks = [

{

'id': 1,

'title': 'Buy Beer',

'description': 'IPA 6 bottles',

'done': False

{

'id': 2,

'title': 'Buy groceries',

'description': 'Beef, Tofu, Sting Onion',

'done': False

}

]

code: app.py

from flask import Flask, jsonify, abort, make_response

from flask_restful import Api, Resource, reqparse, marshal

from flask_httpauth import HTTPBasicAuth

from tasks import tasks, task_fields

from conf import allow_users

app = Flask(__name__, static_url_path="")

api = Api(app)

auth = HTTPBasicAuth()

@auth.get_password

def get_password(username):

if username in allow_users.keys():

return allow_usersusername'password'

return None

@auth.error_handler

def unauthorized():

return make_response(jsonify({'message': 'Unauthorized access'}), 403)

class TaskListAPI(Resource):

decorators = auth.login_required

def __init__(self):

self.reqparse = reqparse.RequestParser()

self.reqparse.add_argument('title', type=str, required=True,

help='No task title provided',

location='json')

self.reqparse.add_argument('description', type=str, default="",

location='json')

super(TaskListAPI, self).__init__()

def get(self):

return {'tasks': marshal(task, task_fields) for task in tasks}

def post(self):

args = self.reqparse.parse_args()

task = {

'id': tasks-1'id' + 1 if len(tasks) > 0 else 1,

'title': args'title',

'description': args'description',

'done': False

}

tasks.append(task)

return {'task': marshal(task, task_fields)}, 201

class TaskAPI(Resource):

decorators = auth.login_required

def __init__(self):

self.reqparse = reqparse.RequestParser()

self.reqparse.add_argument('title', type=str, location='json')

self.reqparse.add_argument('description',

type=str, location='json')

self.reqparse.add_argument('done', type=bool, location='json')

super(TaskAPI, self).__init__()

def get(self, id):

task = [task for task in tasks if task'id' == id]

if len(task) == 0:

abort(404)

return {'task': marshal(task0, task_fields)}

def put(self, id):

task = [task for task in tasks if task'id' == id]

if len(task) == 0:

abort(404)

task = task0

args = self.reqparse.parse_args()

for k, v in args.items():

if v is not None:

taskk = v

return {'task': marshal(task, task_fields)}

def delete(self, id):

task = [task for task in tasks if task'id' == id]

if len(task) == 0:

abort(404)

tasks.remove(task0)

return {'result': True}

api.add_resource(TaskListAPI,

'/todo/api/v3.0/tasks', endpoint='tasks')

api.add_resource(TaskAPI,

'/todo/api/v3.0/tasks/<int:id>', endpoint='task')

if __name__ == '__main__':

app.run(debug=True, port=8080)

TODOWebサービスのAPI version1.0 と比べるとURIのルーティングがスッキリしましたね。

Flask-restx による実装

参考：

Python公式ドキュメント - marshal --- 内部使用向けの Python オブジェクト整列化